package com.xuzz.study.io.mynetty;

import jdk.net.Sockets;
import lombok.Data;

import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @Description
 * @Date 2023/4/5 20:16
 * @Created by xuzz3@asiainfo-sec.com
 */
@Data
public class ThreadSelector extends Thread {

    ThreadSelectorGroup group;
    Selector selector;
    BlockingQueue<Channel> tasks = new LinkedBlockingQueue<>();

    public ThreadSelector(ThreadSelectorGroup group) {
        try {
            this.selector = Selector.open();
            this.group = group;
        }catch (Exception ex){
            ex.printStackTrace();;
        }
    }

    @Override
    public void run() {
        while(true) {
            try {
                int n = this.selector.select();
                if (n > 0) {
                    Set<SelectionKey> selectionKeys = this.selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while(iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        if(key.isAcceptable()) {
                            this.handleAccept(key);
                        } else if (key.isReadable()) {
                            this.handleRead(key);
                        } else if(key.isWritable()){
                            this.handleWrite(key);
                        }
                    }
                } else {
                    this.handleTasks();
                }
            }
            catch (Exception ex) {
                ex.printStackTrace();
            }
        }
    }

    public void handleTasks() {
        if(tasks.isEmpty()) {
            return;
        }
        try
        {
            Channel channel = this.tasks.take();
            if(channel instanceof ServerSocketChannel) {
                ServerSocketChannel serverSocketChannel = (ServerSocketChannel) channel;
                serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
                System.out.println("服务端注册："+Thread.currentThread().getName());
            } else if (channel instanceof SocketChannel) {
                SocketChannel socketChannel = (SocketChannel) channel;
                ByteBuffer byteBuffer = ByteBuffer.allocate(8092);
                socketChannel.register(this.selector,SelectionKey.OP_READ,byteBuffer);
                System.out.println("客户端注册："+Thread.currentThread().getName());
            }
        } catch (Exception e){
            e.printStackTrace();
        }
    }

    public void handleAccept(SelectionKey key) {
        try
        {
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            SocketChannel client = server.accept();

            //注册到workder线程组
            this.group.getBootStrap().workerGroup.next(client).getTasks().put(client);

        } catch (Exception ex){
            ex.printStackTrace();
        }
    }

    public void handleRead(SelectionKey key) {
        try
        {
            SocketChannel channel = (SocketChannel)key.channel();
            ByteBuffer byteBuffer = (ByteBuffer)key.attachment();
            byteBuffer.clear();
            while(true) {
                try
                {
                    int read = channel.read(byteBuffer);
                    if(read>0) {
                        byteBuffer.flip();
                        byte[] data = new byte[byteBuffer.limit()];
                        byteBuffer.get(data);
                        String strData = new String(data);
                        System.out.println("thread:"+Thread.currentThread().getName()+",data:"+strData);
                        channel.register(this.selector,SelectionKey.OP_WRITE,byteBuffer);
                    } else if(read ==0) {
                        break;
                    } else {
                        System.out.println("断开链接");
                        key.cancel();
                        break;
                    }
                }catch (Exception ex){
                    ex.printStackTrace();
                }
            }
        }
        catch (Exception ex){
            ex.printStackTrace();
        }
    }

    public void handleWrite(SelectionKey key) {
        SocketChannel channel = (SocketChannel)key.channel();
        ByteBuffer byteBuffer = (ByteBuffer)key.attachment();
        byteBuffer.flip();
        while(byteBuffer.hasRemaining()) {
            try
            {
                channel.write(byteBuffer);
            }
             catch (Exception ex){
                ex.printStackTrace();;
             }
        }
        try
        {
            byteBuffer.clear();
            channel.register(this.selector,SelectionKey.OP_READ,byteBuffer);
        } catch (Exception e){
            e.printStackTrace();
        }
    }
}
